package com.atguigu.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV1;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.CommonUtil;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author chenlongStart
 * @create 2021-06-25 11:53
 * @desc
 */
public class DwdLogApp extends BaseAppV1 {

    public static void main(String[] args) {
        new DwdLogApp().init(2001, 1, "DwdLogApp", Constant.TOPIC_ODS_LOG, "DwdLogApp");
    }

    public void run(StreamExecutionEnvironment env,DataStreamSource<String> sourceStream) {
        //1、识别新老用户 虽然数据中已经有这个字段，但是不准，需要纠正
        SingleOutputStreamOperator<JSONObject> validatedStream = distinguishNewOrOld(sourceStream);

        //2、流的拆分，利用侧输出流
        Map<String, DataStream<JSONObject>> threeStream = splitStream(validatedStream);

        //3、分别写入到不同的topic
        sendToKafka(threeStream);

    }

    private void sendToKafka(Map<String, DataStream<JSONObject>> threeStream) {
        threeStream.get("start")
                .map(JSONAware::toJSONString)
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_START_LOG));
        threeStream.get("page")
                .map(JSONAware::toJSONString)
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_PAGE_LOG));
        threeStream.get("display")
                .map(JSONAware::toJSONString)
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_DISPLAY_LOG));
    }

    private Map<String, DataStream<JSONObject>> splitStream(SingleOutputStreamOperator<JSONObject> validatedStream) {

        OutputTag<JSONObject> pageTag = new OutputTag<JSONObject>("page") {
        };
        OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>("display") {
        };


        SingleOutputStreamOperator<JSONObject> start = validatedStream
                .process(new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject value,
                                               Context ctx,
                                               Collector<JSONObject> out) throws Exception {
                        if (value.getJSONObject("start") != null) {
                            out.collect(value);
                        } else {
                            if (value.getJSONObject("page") != null) {
                                ctx.output(pageTag, value);
                            }
                            JSONArray displays = value.getJSONArray("displays");
                            if (displays != null) {
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i);
                                    display.put("ts", value.getLong("ts"));
                                    display.put("page_id", value.getJSONObject("page").getString("page_id"));

                                    display.putAll(value.getJSONObject("common"));
                                    ctx.output(displayTag, display);
                                }
                            }
                        }
                    }
                });

        Map<String, DataStream<JSONObject>> map = new HashMap<>();
        map.put("start", start);
        map.put("page", start.getSideOutput(pageTag));
        map.put("display", start.getSideOutput(displayTag));
        return map;
    }

    private SingleOutputStreamOperator<JSONObject> distinguishNewOrOld(DataStreamSource<String> sourceStream) {
        /*
        具体实现逻辑:
        1. 考虑的数据的乱序, 采用事件时间
        2. 窗口
            每个用户的人生中的第一个窗口的中才有可能出现会设置成新用户
                按照事件时间排序, 最小的那个应该是新用户记录, 其他都应该是旧用户

            其他的窗口都应该是旧用户

        3. 如何识别出现某个窗口是否为这个用户的第一个窗口
            使用状态记录
         */
        return sourceStream
                .map(JSON::parseObject)
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((obj, ts) -> obj.getLong("ts")))
                .keyBy(obj -> obj.getJSONObject("common").getString("mid"))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {

                    private ValueState<Long> firstVisitState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        firstVisitState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("firstVisitState", Long.class));
                    }

                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<JSONObject> elements,
                                        Collector<JSONObject> out) throws Exception {
                        // 判断是否为第一个窗口
                        // 是: 按照时间戳排序, 把最小的is_new设置为1其他设置为0
                        // 不是: 所有的数据的si_new都设置为0
                        if (firstVisitState.value() == null) {
                            List<JSONObject> list = CommonUtil.toList(elements);
                            list.sort(Comparator.comparing(o -> o.getLong("ts")));  // 升序
                            for (int i = 0; i < list.size(); i++) {
                                JSONObject common = list.get(i).getJSONObject("common");
                                if (i == 0) {
                                    common.put("is_new", 1);
                                    // 更新状态
                                    firstVisitState.update(list.get(i).getLong("ts"));
                                } else {
                                    common.put("is_new", 0);
                                }
                                // 把每个数据放入collector
                                out.collect(list.get(i));
                            }
                        } else {
                            for (JSONObject obj : elements) {
                                JSONObject common = obj.getJSONObject("common");
                                common.put("is_new", 0);
                                out.collect(obj);
                            }
                        }
                    }
                });
    }
}
